/**
 * Copyright 2018 asiainfo Inc.
 **/
package com.myz.rabbitmqspring.spring.builder;

import com.myz.rabbitmqspring.spring.constants.Constants;
import com.myz.rabbitmqspring.spring.consume.IMessageConsumer;
import com.myz.rabbitmqspring.spring.dto.DetailResponse;
import com.myz.rabbitmqspring.spring.dto.MessageWithTime;
import com.myz.rabbitmqspring.spring.process.MessageProcess;
import com.myz.rabbitmqspring.spring.producer.MessageSender;
import com.myz.rabbitmqspring.spring.retry.RetryCache;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 常量的使用 BuiltinExchangeType
 *
 * @author maoyz on 2018/8/30
 * @version: v1.0
 */
@Component
@Slf4j
public class RabbitMqAccessBuilder {

    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MessagePropertiesConverter messagePropertiesConverter;

    @Autowired
    private Jackson2JsonMessageConverter jsonMessageConverter;

    public RabbitMqAccessBuilder() {
    }

    public RabbitMqAccessBuilder(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    /**
     * 创建direct,默认
     *
     * @param exchange
     * @param routingKey
     * @param queue
     * @return
     * @throws IOException
     */
    public MessageSender buildMessageSender(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSender(exchange, routingKey, queue, BuiltinExchangeType.DIRECT.getType());
    }

    /**
     * 创建topic
     *
     * @param exchange
     * @param routingKey
     * @return
     * @throws IOException
     */
    public MessageSender buildTopicMessageSender(final String exchange, final String routingKey) throws IOException {
        return buildMessageSender(exchange, routingKey, null, BuiltinExchangeType.TOPIC.getType());
    }

    /**
     * 1 构造template, exchange, routingkey等
     * 2 设置message序列化方法
     * 3 设置发送确认
     * 4 构造sender方法
     */
    public MessageSender buildMessageSender(final String queue, final String exchange, final String routingKey, final String type) throws IOException {
        // connection
        final Connection connection = connectionFactory.createConnection();

        // 绑定交换机和队列
        if (BuiltinExchangeType.DIRECT.getType().equals(type)) {
            buildQueue(exchange, routingKey, queue, connection, BuiltinExchangeType.TOPIC.getType());
        } else if (BuiltinExchangeType.TOPIC.getType().equals(type)) {
            buildTopic(exchange, connection);
        }

        RetryCache retryCache = new RetryCache();

        return new MessageSender() {

            // 设置
            {
                retryCache.setMessageSender(this);
            }

            @Override
            public DetailResponse send(Object message) {
                Long id = retryCache.generateId();
                long timeMillis = System.currentTimeMillis();
                return send(new MessageWithTime(id, timeMillis, message));
            }

            @Override
            public DetailResponse send(MessageWithTime messageWithTime) {
                retryCache.add(messageWithTime);

                /*
                 * 构建Message ,主要是使用 msgId 将 message 和 CorrelationData 关联起来。
                 * 这样当消息发送到交换机失败的时候，在 ConfirmCallBack 中就可以根据
                 * correlationData.getId()即 msgId,知道具体是哪个message发送失败,进而进行处理。
                 * 将 msgId和 message绑定
                 */
                Message message = MessageBuilder.withBody(((String) messageWithTime.getMessage()).getBytes())
                        .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                        .setCorrelationId(String.valueOf(messageWithTime.getId()))
                        .build();

                Object receive = null;

                try {
                    receive = rabbitTemplate.convertSendAndReceive(message, new CorrelationData(String.valueOf(messageWithTime.getId())));
                } catch (AmqpException e) {
                    log.info("send failed :" + e.getMessage());
                    try {
                        // 失败，重试
                        receive = rabbitTemplate.convertSendAndReceive(message, new CorrelationData(String.valueOf(messageWithTime.getId())));
                    } catch (AmqpException e1) {
                        log.error("send failed again :" + e.getMessage());
                        return new DetailResponse(Boolean.TRUE, e.getMessage());
                    }
                }
                log.info("<===================== send success ====================>");
                // 成功发送消息
                return new DetailResponse(Boolean.TRUE, (String) receive);
            }
        };
    }


    /**
     * 创建direct 消费者
     *
     * @param exchange
     * @param routingKey
     * @param queue
     * @param messageProcess
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> IMessageConsumer buildMessageConsumer(String exchange, String routingKey, final String queue, final MessageProcess<T> messageProcess) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, BuiltinExchangeType.DIRECT.getType());
    }

    /**
     * 创建topic 消费者
     *
     * @param exchange
     * @param routingKey
     * @param queue
     * @param messageProcess
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> IMessageConsumer buildTopicMessageConsumer(String exchange, String routingKey, final String queue, final MessageProcess<T> messageProcess) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, BuiltinExchangeType.TOPIC.getType());
    }

    /**
     * 1 创建连接和channel
     * 2 设置message序列化方法
     * 3 consume
     */
    public <T> IMessageConsumer buildMessageConsumer(final String queue, final String exchange, final String routingKey, final MessageProcess<T> messageProcess, final String type) throws IOException {

        final Connection connection = connectionFactory.createConnection();

        buildQueue(exchange, routingKey, queue, connection, type);

        return new IMessageConsumer() {
            Channel channel;

            {
                connection.createChannel(false);
            }

            /**
             * 1 通过delivery获取原始数据
             * 2 将原始数据转换为特定类型的包
             * 3 处理数据
             * 4 手动发送ack确认
             */
            @Override
            public DetailResponse consume() {
                try {
                    GetResponse response = channel.basicGet(queue, false);

                    while (response == null) {
                        response = channel.basicGet(queue, false);
                        TimeUnit.SECONDS.sleep(Constants.ONE_SECOND);
                    }

                    // 构建message
                    Message message = MessageBuilder.withBody(response.getBody())
                            .andProperties(messagePropertiesConverter.toMessageProperties(response.getProps(), response.getEnvelope(), StandardCharsets.UTF_8.displayName()))
                            .build();

                    @SuppressWarnings("unchecked")
                    T messageBean = (T) jsonMessageConverter.fromMessage(message);

                    DetailResponse process = null;
                    try {
                        process = messageProcess.process(messageBean);
                    } catch (Exception e) {
                        log.error("************** messageProcess exception", e);
                        process = new DetailResponse(false, "process exception: " + e);
                    }

                    if (process.isSuccess()) {
                        // 消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
                        channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
                    } else {
                        // ack返回false，并重新回到队列
                        TimeUnit.MICROSECONDS.sleep(200);
                        log.error("process message failed: " + process.getErrorMessage());
                        channel.basicNack(response.getEnvelope().getDeliveryTag(), false, true);
                    }

                    return process;
                } catch (Exception e) {
                    log.error("************** messageProcess exception", e);
                    return new DetailResponse(Boolean.FALSE, e.toString());
                }
            }
        };
    }

    /**
     * 创建队列
     *
     * @param exchange
     * @param routingKey
     * @param queue
     * @param connection
     * @param type
     */
    private void buildQueue(String exchange, String routingKey, String queue, Connection connection, String type) throws IOException {
        Channel channel = connection.createChannel(false);

        // 绑定相应交换机
        if (BuiltinExchangeType.DIRECT.getType().equals(type)) {
            channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT, true, false, null);
        } else if (BuiltinExchangeType.TOPIC.getType().equals(type)) {
            channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true, false, null);
        }

        // 队列声明queueDeclare
        channel.queueDeclare(queue, true, false, false, null);
        // 绑定路由
        channel.queueBind(queue, exchange, routingKey);

        try {
            channel.close();
        } catch (TimeoutException e) {
            log.error("close channel time out ", e);
        }
    }

    private void buildTopic(String exchange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true, false, null);
    }

}
